package org.ourgrid.acceptance.broker;

import static org.junit.Assert.assertTrue;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.Map.Entry;

import org.easymock.classextension.EasyMock;
import org.junit.After;
import org.junit.Test;
import org.ourgrid.acceptance.util.broker.Req_302_Util;
import org.ourgrid.acceptance.util.broker.Req_304_Util;
import org.ourgrid.acceptance.util.broker.Req_309_Util;
import org.ourgrid.acceptance.util.broker.Req_311_Util;
import org.ourgrid.acceptance.util.broker.Req_312_Util;
import org.ourgrid.acceptance.util.broker.Req_314_Util;
import org.ourgrid.acceptance.util.broker.Req_327_Util;
import org.ourgrid.acceptance.util.broker.Req_328_Util;
import org.ourgrid.acceptance.util.broker.Req_329_Util;
import org.ourgrid.acceptance.util.broker.TestJob;
import org.ourgrid.broker.BrokerComponent;
import org.ourgrid.broker.BrokerConfiguration;
import org.ourgrid.broker.actions.ErrorOcurredMessageHandle;
import org.ourgrid.broker.actions.HereIsGridProcessResultMessageHandle;
import org.ourgrid.broker.actions.WorkerIsReadyMessageHandle;
import org.ourgrid.broker.commands.CommandExecutor;
import org.ourgrid.broker.commands.SchedulerData;
import org.ourgrid.broker.dao.JobDAO;
import org.ourgrid.broker.dao.WorkerEntry;
import org.ourgrid.broker.scheduler.SchedulerIF;
import org.ourgrid.broker.scheduler.workqueue.JobInfo;
import org.ourgrid.common.executor.ExecutorResult;
import org.ourgrid.common.interfaces.LocalWorkerProvider;
import org.ourgrid.common.interfaces.Worker;
import org.ourgrid.common.interfaces.WorkerClient;
import org.ourgrid.common.interfaces.control.BrokerControl;
import org.ourgrid.common.interfaces.control.BrokerControlClient;
import org.ourgrid.common.interfaces.to.GridProcessErrorTypes;
import org.ourgrid.common.interfaces.to.GridProcessState;
import org.ourgrid.common.job.GridProcess;
import org.ourgrid.common.job.Job;
import org.ourgrid.common.job.Task;
import org.ourgrid.common.spec.job.IOBlock;
import org.ourgrid.common.spec.job.IOEntry;
import org.ourgrid.common.spec.job.JobSpec;
import org.ourgrid.common.spec.job.TaskSpec;
import org.ourgrid.common.spec.peer.PeerSpec;
import org.ourgrid.common.spec.worker.WorkerSpec;
import org.ourgrid.peer.PeerComponentContextFactory;
import org.ourgrid.worker.controller.GridProcessError;

import br.edu.ufcg.lsd.commune.container.ObjectDeployment;
import br.edu.ufcg.lsd.commune.container.contextfactory.PropertiesFileParser;
import br.edu.ufcg.lsd.commune.functionaltests.util.TestContext;
import br.edu.ufcg.lsd.commune.functionaltests.util.TestStub;
import br.edu.ufcg.lsd.commune.identification.DeploymentID;
import br.edu.ufcg.lsd.commune.identification.ServiceID;
import br.edu.ufcg.lsd.commune.test.AcceptanceTestUtil;

public class TestManyJob extends BrokerAcceptanceTestCase {
	
	@After
    public void tearDown() throws Exception {
		createComponentContext("1", "1");
		super.tearDown();
    }
	
	@Test public void test_Replicaton_ManyWorkers() throws Exception {
		
		
		final String maxReplicas = "2";
		
		TestContext context = createComponentContext(maxReplicas);
		
		Req_302_Util req_302_Util = new Req_302_Util(context);
		Req_314_Util req_314_Util = new Req_314_Util(context);
		Req_329_Util req_329_Util = new Req_329_Util(context);
		
		//start broker
		BrokerComponent broker = req_302_Util.startBroker();
		
		//adding jobs
		List<TestStub> workerStubs = prepareTestObjetcts(broker, context, 5);
		
		WorkerClient workerClient = req_314_Util.getWorkerClient(broker);
		ObjectDeployment wcOD = req_329_Util.getWorkerClientDeployment(broker);
		
		//sending workerIsReady message
		for (TestStub workerStub : workerStubs) {
			AcceptanceTestUtil.setExecutionContext(broker, wcOD, workerStub.getDeploymentID());
			workerClient.sendMessage(new WorkerIsReadyMessageHandle());
		}
		
		Thread.sleep(15000);
		
		Job job1 = getJobInfo().getJob(1);
		Job job2 = getJobInfo().getJob(1);
		Job job3 = getJobInfo().getJob(1);
		Job job4 = getJobInfo().getJob(1);
		Job job5 = getJobInfo().getJob(1);
		
		
		
		
		
		/*List<WorkerEntry> workers = getTaskWorkers(1, 1, job, GridProcessState.RUNNING);
		assertTrue(workers.size() == 1);
		
		
		Thread.sleep(15000);
		
		job = getJobInfo().getJob(1);

		workers = getTaskWorkers(1, 1, job, GridProcessState.RUNNING);
		assertTrue(workers.size() == 0);
		
		workers = getTaskWorkers(1, 2, job, GridProcessState.RUNNING);
		assertTrue(workers.size() == 0);
		
		workers = getTaskWorkers(1, 3, job, GridProcessState.RUNNING);
		assertTrue(workers.size() == 0);
		
		workers = getTaskWorkers(1, 4, job, GridProcessState.RUNNING);
		assertTrue(workers.size() == 0);
		
		workers = getTaskWorkers(1, 5, job, GridProcessState.RUNNING);
		assertTrue(workers.size() == 0);
		
		workers = getTaskWorkers(1, 1, job, GridProcessState.FINISHED);
		assertTrue(workers.size() == 1);
		
		workers = getTaskWorkers(1, 2, job, GridProcessState.FINISHED);
		assertTrue(workers.size() == 1);
		
		workers = getTaskWorkers(1, 3, job, GridProcessState.FINISHED);
		assertTrue(workers.size() == 1);
		
		workers = getTaskWorkers(1, 4, job, GridProcessState.FINISHED);
		assertTrue(workers.size() == 1);
		
		workers = getTaskWorkers(1, 5, job, GridProcessState.FINISHED);
		assertTrue(workers.size() == 1);*/
	}
	
	
	private DeploymentID getWorkerID(List<TestStub> stubs, ServiceID serviceID) {
		
		String login = serviceID.getUserName() + "@" + serviceID.getServerName();
		
		ServiceID workerServiceID = null;
		for (TestStub testStub : stubs) {
			workerServiceID = testStub.getDeploymentID().getServiceID();
			
			String workerLogin = workerServiceID.getUserName() + "@" + workerServiceID.getServerName();
			if (login.equals(workerLogin)) {
				return testStub.getDeploymentID();
			}
		}
		
		return null;
	}
	
	private List<TestStub> prepareTestObjetcts(BrokerComponent broker, TestContext context, int jobs) 
		throws Exception {
		
		Req_309_Util req_309_Util = new Req_309_Util(context);
		Req_311_Util req_311_Util = new Req_311_Util(context);
		Req_327_Util req_327_Util = new Req_327_Util(context);
		Req_328_Util req_328_Util = new Req_328_Util(context);
		Req_312_Util req_312_Util = new Req_312_Util(context);
		Req_329_Util req_329_Util = new Req_329_Util(context);
		
		//set peer
		PeerSpec peer = req_309_Util.setPeers("username", "servertest", "publickey1", broker);
		
		//call doNotifyRecovery passing a peer with username test
		DeploymentID deploymentID = req_328_Util.createPeerDeploymentID("publicKey1", peer);
		TestStub peerTestStub = req_327_Util.notifyPeerRecovery(peer, deploymentID, broker);
		
		//do login with peer
		req_311_Util.verifyLogin(broker, "publickey1", false, false, null, peerTestStub);
		
		List<TestJob> testJobs = prepareJobs(jobs, broker, context, peerTestStub, true);
		
		//Receveing workers
		List<TestStub> workerStubs = new ArrayList<TestStub>();
		
		TestStub workerTestStub = null;
		WorkerSpec workerSpec = null;
		Map<String, String> attributes = null;
		
		for (int i = 1; i <= jobs; i++) {
			
			attributes = new HashMap<String, String>();
			attributes.put(WorkerSpec.ATT_MEM, "mem = 256");
			
			workerSpec = new WorkerSpec(attributes);
			workerSpec.putAttribute(WorkerSpec.ATT_USERNAME, "worker" + i);
			workerSpec.putAttribute(WorkerSpec.ATT_SERVERNAME, "ortho.ece.ubc.ca");
			
			TestJob testJob = testJobs.get(i-1);
			
			workerTestStub = req_312_Util.receiveWorker(broker, "workerPublicKey" + i, true, true, true, true, 
					workerSpec, "publickey1", peerTestStub, testJob);
			
			workerStubs.add(workerTestStub);
		}
		
		//Scheduling
		EasyMock.reset(broker.getLogger());
		
		for (TestStub testStub : workerStubs) {
			Worker worker = (Worker) testStub.getObject();
			EasyMock.reset(worker);
		}
		
		EasyMock.reset(broker.getLogger());
		
		Set<SchedulerIF> schedulers = broker.getDAO(JobDAO.class).getSchedulers();
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		for (SchedulerIF schedulerIF : schedulers) {
			outputCommands.addAll(schedulerIF.schedule());
		}
		
		ObjectDeployment od = req_329_Util.getBrokerControlDeployment(broker);
		CommandExecutor.getInstance().execute(outputCommands, broker.createServiceManager(od));
		
		return workerStubs;
	}
	
	private List<TestJob> prepareJobs(int jobs, BrokerComponent broker, TestContext context, 
			TestStub peerTestStub, boolean hasGetFiles) 
		throws Exception {
		
		Req_304_Util req_304_Util = new Req_304_Util(context);
		Req_314_Util req_314_Util = new Req_314_Util(context);
		
		//Job creation
		List<TaskSpec> taskSpecs = new ArrayList<TaskSpec>();
		for (int i = 1; i <= 1; i++) {
			
			TaskSpec taskSpec = null;
			if (hasGetFiles) {
				IOBlock finalBlock = new IOBlock();
				finalBlock.putEntry(new IOEntry("GET", "remoteFile1.txt", "localFile1.txt"));
				finalBlock.putEntry(new IOEntry("GET", "remoteFile2.txt", "localFile2.txt"));
				taskSpec = new TaskSpec(null, "echo", finalBlock, "echo");
			} else {
				taskSpec = new TaskSpec(null, "echo", null, "echo");
			}
			
			taskSpec.setSourceDirPath(BrokerAcceptanceTestCase.BROKER_TEST_DIR);
			taskSpecs.add(taskSpec);
		}
		
		BrokerControl brokerControl = req_304_Util.getBrokerControl(broker);
		ObjectDeployment brokerOD = req_314_Util.getBrokerControlDeployment(broker);
		BrokerControlClient brokerControlClientMock = EasyMock.createNiceMock(BrokerControlClient.class);
		
		LocalWorkerProvider lwp = (LocalWorkerProvider) peerTestStub.getObject();
		EasyMock.reset(lwp);
		EasyMock.reset(broker.getLogger());
		
		AcceptanceTestUtil.setExecutionContext(broker, brokerOD, brokerOD.getDeploymentID().getPublicKey());
		
		List<TestJob> testJobs = new ArrayList<TestJob>();
		
		for (int i = 1; i <= jobs; i++) {
			JobSpec jobSpec = new JobSpec("Test Job", "mem = 256", taskSpecs);
			brokerControl.addJob(brokerControlClientMock, jobSpec);
			Job job = getJobInfo().getJob(i);
			testJobs.add(new TestJob(job, jobSpec));
		}
		
		return testJobs;
	}
	
	
	private List<WorkerEntry> getTaskWorkers(int jobID, int taskID, Job job, GridProcessState state) {
		List<WorkerEntry> taskWorkers = new ArrayList<WorkerEntry>();
		
		for (Task task : job.getTasks()) {
			if (task.getJobId() == jobID && task.getTaskid() == taskID && task.getState().equals(state)) {
				for (GridProcess process : task.getGridProcesses()) {
					if (process.getState().equals(state)) {
						taskWorkers.add(process.getWorkerEntry());
					}	
				}
			}	
		}
		
		return taskWorkers;
	}
	
	protected TestContext createComponentContext(String maxReplicas, String maxBLFails, String maxFails) {
		TestContext testContext = new TestContext(
				new PeerComponentContextFactory(
						new PropertiesFileParser(BrokerAcceptanceTestCase.PROPERTIES_FILENAME
						)).createContext());
		
		testContext.set(BrokerConfiguration.PROP_MAX_REPLICAS, maxReplicas);
		testContext.set(BrokerConfiguration.PROP_MAX_BL_FAILS, maxBLFails);
		testContext.set(BrokerConfiguration.PROP_MAX_FAILS, maxFails);
		
		Properties properties = new Properties();
		for (Entry<String, String> entry : testContext.getProperties().entrySet()) {
			properties.setProperty(entry.getKey(), entry.getValue());
		}
		
		try {
			properties.store(new FileOutputStream(BrokerAcceptanceTestCase.PROPERTIES_FILENAME), null);
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		return testContext;
	}
	
	
	
	protected TestContext createComponentContext(String maxReplicas) {
		return createComponentContext(maxReplicas, "1", "3");
	}
	
	protected TestContext createComponentContext(String maxReplicas, String maxBLFails) {
		return createComponentContext(maxReplicas, maxBLFails, "3");
	}
	
	private JobInfo getJobInfo() {
		return JobInfo.getInstance();
	}
	
}
